package com.neuedu.heart

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * 计算出心脏病患者中吸烟、饮酒、锻炼 习惯的比例
 */
// 定义封装的对象
case class HabitInCardioPercentObject(smoke:Double,alco:Double,exercise:Double)
object HabitInCardioPercent {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","root")
    //-:Spark配置对象
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
    //-:SparkSession 对象
    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .appName(getClass.getName)
      .getOrCreate()
    //-:读取清洗之后的数据
    val df = sparkSession.read.table("db_heart.dwd_cardio")

    //-:获取烟民的数量，有心血管疾病
    //-：烟民的数量
    val smokeDs = df.filter("smoke = 1")
    //-：烟民中有心血管疾病的数量
    val smokeCardioDs = df.filter("smoke = 1")
      .filter("cardio = 1")
    //: 患有心血管的烟民在整个烟民中所占的比例
    val cardioInSmokePrecent = rate(smokeCardioDs.count().toDouble/smokeDs.count().toDouble)

    //-:获取饮酒的数量，有心血管疾病
    //-:有饮酒习惯的人数量
    val alcoDs = df.filter("alco = 1")
    //-有饮酒习惯的人并且有心血管疾病的数量
    val alcoCardioDs = alcoDs.filter("cardio = 1")
    //:患有心血管的饮酒者在整个饮酒者中所占的比例
    val cardioInAlcoPrecent = rate(alcoCardioDs.count().toDouble/alcoDs.count().toDouble)

    //-:健身习惯的人
    val activeDs = df.filter("active = 1")
    //-:健身习惯人中患有心血管疾病的人数
    val activeCardioDs = activeDs.filter("cardio = 1")
    //-:健身习惯人中患有心血管疾病的人数 与 健身习惯的人
    val cardioInActivePrecent = rate(activeCardioDs.count().toDouble/activeDs.count().toDouble)

    //-:使用构造器封装将数据封装成对象，然后保存到队列中，然后在转换为 dataset
    //-：定义封装的对象（在下面定义）
    val habitInCardioPercent = HabitInCardioPercentObject(cardioInSmokePrecent,cardioInAlcoPrecent,cardioInActivePrecent)
    val seq = Seq(habitInCardioPercent)
    import sparkSession.implicits._
    val ds = seq.toDS()

    //-:将数据保存到hive中
    ds.write.mode(SaveMode.Append).saveAsTable("db_heart.dws_cardio_habit")

    //-资源的释放
    sparkSession.stop()
    sparkSession.close()
  }
  // 定义百分比转换的函数
  def rate(number: Double): Double = {
    BigDecimal(number).setScale(4,BigDecimal.RoundingMode.HALF_UP).toDouble * 100
  }
}
